package homework;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class MaxTemperature {
    public static void main(String[] args) throws Exception {
        // if (args.length != 2) {
        // System.err.println("Usage:MaxTEmperature");
        // System.exit(-1);
        // }
        Configuration conf = new Configuration();
        // conf.set("HADOOP_USER", "hadoop");
        Job job = Job.getInstance(conf);

        job.setJarByClass(MaxTemperature.class);
        job.setJobName("Max temperature");

        job.setMapperClass(MaxTemperatureMapper.class);
        job.setReducerClass(MaxTemperatureReducer.class);
        job.setCombinerClass(MaxTemperatureReducer.class);// 合并
        // job.setPartitionerClass(TemperaturePartitioner.class);// 分区

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // job.setNumReduceTasks(2);

        FileInputFormat.addInputPath(job, new Path("hdfs://192.168.3.101:9000/weatherAnalyse/input/cndcdata.txt"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.3.101:9000/weatherAnalyse/output/result"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class MaxTemperatureMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        private static final int MISSING = 9999;

        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String yearAndmonth = line.substring(15, 21);// 16-23 20170101
            int airTemperature;
            if (line.charAt(87) == '+') {// 截取气温的符号
                airTemperature = Integer.parseInt(line.substring(88, 92));// 气温是正数，不用获得“+”

            } else {
                airTemperature = Integer.parseInt(line.substring(87, 92));// 气温是负数，直接将“-”加入到数据中
            }
            String quality = line.substring(92, 93);
            // 分析空气温度数值是否合法
            if (airTemperature != MISSING && quality.matches("[01459]")) {
                context.write(new Text(yearAndmonth), new IntWritable(airTemperature));
            }

        }
    }

    public static class MaxTemperatureReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int maxValue = Integer.MIN_VALUE;
            for (IntWritable value : values) {
                maxValue = Math.max(maxValue, value.get());
            }
            context.write(key, new IntWritable(maxValue));

        }
    }

    public static class TemperaturePartitioner<K, V> extends Partitioner<K, V> {

        @Override
        // 自定义partition的数量需要和reduce
        // task数量保持一致，在setNumReduceTasks是数量填为2，因为我们要分成两个文件
        public int getPartition(K key, V value, int numPartitions) {
            String monthStr = key.toString().substring(4, 6);// 从年月中截取月份
            int month = Integer.parseInt(monthStr);
            if (month > 6) {
                return 0;
            } else {
                return 1;
            }
        }
    }

}
